package com.heima.kafka.simple;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

/**
 * 消费者，监听消息队列，获取消息内容
 */
public class ConsumerStart {
    public static void main(String[] args) {
        //2.设置连接kafka的信息
        Properties properties=new Properties();
        //2.1 设置kafka地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.105:9092");
        //2.2设置对key,value进行字符串反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        //加入一个组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"group3");

        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//手动提交偏移量
        //1.创建消费者对象
        KafkaConsumer<String,String > kafkaConsumer=new KafkaConsumer<String, String>(properties);
        //3.订阅主题(监听的是哪个队列)
        kafkaConsumer.subscribe(Collections.singletonList("kafka-stream-consumer"));

        //4.拉取消息内容
        while (true){
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("消息的key:"+record.key()+"消息的value:"+record.value());
                /*System.out.println("消息的key:"+record.key()+",消息的内容:"+record.value()+",消息所在的分区:"+record.partition()+",消息所在的序列号(偏移量):"+record.offset());
               // kafkaConsumer.commitSync();//同步提交

                //异步提交
                kafkaConsumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if(exception!=null){
                            //进行重试提交
                            System.out.println(exception);
                        }
                    }
                });*/
             }
        }

    }
}
